package com.bw.yk01;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import scala.Tuple4;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/*
1,192.168.0.1,fail,1558430842
1,192.168.0.2,fail,1558430843
1,192.168.0.2,fail,1558430844
1,192.168.0.3,fail,1558430850
1,192.168.0.3,fail,1558430851
2,192.168.10.10,fail,1558430846
2,192.168.10.10,success,1558430848
2,192.168.10.10,fail,1558430850
 */
public class FlinkTM5_3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv);


        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "group1");
        DataStream<String> stream = fsEnv.addSource(new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties));



        SingleOutputStreamOperator<LoginBean> mapDS = stream.map(new MapFunction<String, LoginBean>() {
            @Override
            public LoginBean map(String s) throws Exception {
                String[] li = s.split(",");
                return new LoginBean(li[0], li[1], li[2], Long.valueOf(li[3].trim()) * 1000);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<LoginBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((event, timestamp) -> event.getTs()));

        //API方式CEP=========================================
        //1、确定输入
        KeyedStream<LoginBean, String> input = mapDS.keyBy(new KeySelector<LoginBean, String>() {
            @Override
            public String getKey(LoginBean loginBean) throws Exception {
                return loginBean.getUid();
            }
        });
        //2、制定规则
        Pattern<LoginBean, LoginBean> pattern = Pattern.<LoginBean>begin("one").where(new SimpleCondition<LoginBean>() {
                    public boolean filter(LoginBean value) throws Exception {
                        return value.getStatus().equals("fail");
                    }
                })
                .times(2)
                .consecutive()
                .within(Time.seconds(5));
        //3、讲规则作用于流
        PatternStream<LoginBean> patternStream = CEP.pattern(input, pattern);
        //4、提取数据
        DataStream<String> alerts = patternStream.select(new PatternSelectFunction<LoginBean, String>() {
            @Override
            public String select(Map<String, List<LoginBean>> map) throws Exception {
                return "连续失败2次"+map;
            }
        });
        alerts.print();



        //fsTableEnv.createTemporaryView("t_cep",mapDS);
        //fsTableEnv.sqlQuery("select * from t_cep").execute().print();




        fsEnv.execute();

    }
}
